package me.zhengjie.gas.mqtt;

import cn.hutool.core.lang.UUID;
import io.vertx.core.Vertx;
import io.vertx.mqtt.MqttClient;
import io.vertx.mqtt.MqttClientOptions;
import lombok.extern.slf4j.Slf4j;
import me.zhengjie.gas.mqtt.config.MqttClientProperties;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
/**
 * @author tsg
 * @version 1.0
 * @description: MqttClientFactory
 * @date 2022/4/28 10:24
 */
@Slf4j
public class MqttClientFactory extends BasePooledObjectFactory<MqttClient> {

    private MqttClientProperties config;

    private Vertx vertx = Vertx.vertx();

    public MqttClientFactory(MqttClientProperties config) {
        this.config = config;
    }

    /**
     * 创建client对象
     */
    @Override
    public MqttClient create() throws InterruptedException {
        MqttClientOptions options = new MqttClientOptions();
        options.setMaxMessageSize(config.getMaxMessageSize());
        options.setPassword(config.getPassword());
        options.setUsername(config.getUsername());
        options.setAckTimeout(config.getAckTimeout());
        options.setAutoGeneratedClientId(config.isAutoGeneratedClientId());
        options.setClientId(config.getClientId() + UUID.randomUUID().toString());
        options.setAutoKeepAlive(config.isAutoKeepAlive());
        options.setCleanSession(config.isCleanSession());
        options.setKeepAliveInterval(config.getKeepAliveInterval());
        options.setWillFlag(config.isWillFlag());
        options.setWillQoS(config.getWillQoS());
        MqttClient client = MqttClient.create(Vertx.vertx(), options);
        MqttClient connect = client.connect(config.getPort(), config.getHost(), r -> {
            if (r.succeeded()) {
                log.info("{}:连接成功回调",log.getName());
            }
            if (r.failed()) {
                log.error("{}:连接失败回调",log.getName());
            }
        });
        setupReconnect(connect);

        return connect;
    }

    private void setupReconnect(MqttClient client) {
        client.closeHandler(v -> {
            log.error("{}:连接被断开, 计划重连!", log.getName());
            vertx.setTimer(5000, id -> {
                try {
                    client.disconnect(); // 确保客户端处于断开状态再重连
                    client.connect(config.getPort(), config.getHost(), r -> {
                        if (r.succeeded()) {
                            log.info("{}:连接成功回调",log.getName());
                        }
                        if (r.failed()) {
                            log.error("{}:连接失败回调",log.getName());
                        }
                    });
                } catch (Exception e) {
                    log.error("Error during reconnect setup: {}", e.getMessage());
                }
            });
        });
    }


    /**
     * 用PooledObject封装对象放入池中
     */
    @Override
    public PooledObject<MqttClient> wrap(MqttClient client) {
        return new DefaultPooledObject<>(client);
    }
    /**
     * 销毁client对象
     */
    @Override
    public void destroyObject(PooledObject<MqttClient> pooled) {
        if (pooled == null) {
            return;
        }
        MqttClient client = pooled.getObject();
        if (client.isConnected()) {
            client.disconnect();
        }
    }
    /**
     * 验证client对象,创建连接客户端是异步的,需要在获取的使用对客户端进行判断
     */
    @Override
    public boolean validateObject(PooledObject<MqttClient> pooled) {
        MqttClient client = pooled.getObject();
        boolean flag = false;
        while (!flag){
            flag = client.isConnected();
        }
        return flag;
    }
}

